package com.xxl.job.consume.kafka;

import com.xxl.job.consume.model.*;

import com.xxl.job.consume.service.AlarmJudgeService;

import com.xxl.job.consume.utils.JSONUtils;
import com.xxl.job.consume.utils.ThreadLocalDateUtil;
import com.xxl.job.consume.utils.TransUtils;
import com.xxl.job.core.log.XxlJobLogger;
import com.xxl.job.executor.model.CurrentInfoMjModel;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.Date;
import java.util.List;

/**
 * kafka消费端
 */
@Service
public class KafkaConsumeService {

    @Autowired
    private AlarmJudgeService alarmJudgeService;
//    @Autowired
//    private SignalDefService signalDefService;
//    @Autowired
//    private CurrentInfoRepository currentInfoRepository;
//    @Autowired
//    private KafkaProduceService kafkaProduceService;

    /**
     * 处理环控
     * @param record
     */
    public void dealHkMessage(ConsumerRecord record){
        try {
            CurrentInfoHkModel currentInfoHkModel= JSONUtils.json2pojo(record.value().toString(), CurrentInfoHkModel.class);
            alarmJudgeService.alarmAssemble(currentInfoHkModel.getDevCode(),currentInfoHkModel.getSignalCode(),currentInfoHkModel.getSignalType()
                   , currentInfoHkModel.getSignalName()  ,currentInfoHkModel.getGatherId(),currentInfoHkModel.getValue(),currentInfoHkModel.getValunit(),
                    currentInfoHkModel.getIsSaveEs());
        } catch (IOException e) {
            e.printStackTrace();
            XxlJobLogger.log("kafka消费端环控实体{}转换失败！",record.value().toString(),e.getMessage());
        }
    }




    /**
     * 处理入侵
     * @param record
     */
    public void dealRqMessage(ConsumerRecord record){
        try {
            CurrentInfoRqModel currentInfoRqModel= JSONUtils.json2pojo(record.value().toString(), CurrentInfoRqModel.class);
            alarmJudgeService.alarmAssemble(currentInfoRqModel.getDevCode(),currentInfoRqModel.getSignalCode(),currentInfoRqModel.getSignalType()
                    , currentInfoRqModel.getSignalName()  ,currentInfoRqModel.getGatherId(),currentInfoRqModel.getValue(),currentInfoRqModel.getValunit(),"0");
        } catch (IOException e) {
            e.printStackTrace();
            XxlJobLogger.log("kafka消费端入侵实体{}转换失败！",record.value().toString(),e.getMessage());
        }
    }

    /**
     * 处理电话
     * @param record
     */
    public void dealDhMessage(ConsumerRecord record){
        try {

            CurrentInfoDhModel currentInfoDhModel= JSONUtils.json2pojo(record.value().toString(), CurrentInfoDhModel.class);

            alarmJudgeService.alarmAssemble(currentInfoDhModel.getDevCode(),currentInfoDhModel.getSignalCode(),currentInfoDhModel.getSignalType()
                    , currentInfoDhModel.getSignalName()  ,currentInfoDhModel.getGatherId(),currentInfoDhModel.getValue(),currentInfoDhModel.getValunit(),"0");
        } catch (IOException e) {
            e.printStackTrace();
            XxlJobLogger.log("kafka消费端电话实体{}转换失败！",record.value().toString(),e.getMessage());
        }
    }

    /**
     * 处理门禁
     * @param record
     */
    public void dealMjMessage(ConsumerRecord record){
        try {

            //主要是value的状态  0关门 1开门
            CurrentInfoMjModel currentInfoMjModel= JSONUtils.json2pojo(record.value().toString(), CurrentInfoMjModel.class);
            alarmJudgeService.alarmAssemble(currentInfoMjModel.getDevCode(),currentInfoMjModel.getSignalCode(),currentInfoMjModel.getSignalType()
                    , currentInfoMjModel.getSignalName()  ,currentInfoMjModel.getGatherId(),currentInfoMjModel.getValue(),currentInfoMjModel.getValunit(),"0");
        } catch (IOException e) {
            e.printStackTrace();
            XxlJobLogger.log("kafka消费端门禁实体{}转换失败！",record.value().toString(),e.getMessage());
        }
    }


    /**
     * 处理消防
     * @param record
     */
    public void dealXfMessage(ConsumerRecord record){
        try {
            CurrentInfoXfModel currentInfoXfModel= JSONUtils.json2pojo(record.value().toString(), CurrentInfoXfModel.class);
            alarmJudgeService.alarmAssemble(currentInfoXfModel.getDevCode(),currentInfoXfModel.getSignalCode(),currentInfoXfModel.getSignalType()
                    , currentInfoXfModel.getSignalName()  ,currentInfoXfModel.getGatherId(),currentInfoXfModel.getValue(),currentInfoXfModel.getValunit(),"0");
        } catch (IOException e) {
            e.printStackTrace();
            XxlJobLogger.log("kafka消费端消防实体{}转换失败！",record.value().toString(),e.getMessage());
        }
    }

    /**
     * 处理AP
     * @param record
     */
    public void dealApMessage(ConsumerRecord record){
        try {
            CurrentInfoApModel currentInfoApModel= JSONUtils.json2pojo(record.value().toString(), CurrentInfoApModel.class);
            alarmJudgeService.alarmAssemble(currentInfoApModel.getDevCode(),currentInfoApModel.getSignalCode(),currentInfoApModel.getSignalType()
                    , currentInfoApModel.getSignalName()  ,currentInfoApModel.getGatherId(),currentInfoApModel.getValue(),currentInfoApModel.getValunit(),"0");
        } catch (IOException e) {
            e.printStackTrace();
            XxlJobLogger.log("kafka消费端AP实体{}转换失败！",record.value().toString(),e.getMessage());
        }
    }

    /**
     * 处理采集口
     * @param record
     */
    public void dealGatherMessage(ConsumerRecord record){
        try {
            CurrentInfoGatherModel currentInfoGatherModel= JSONUtils.json2pojo(record.value().toString(), CurrentInfoGatherModel.class);
            alarmJudgeService.alarmAssemble(currentInfoGatherModel.getDevCode(),currentInfoGatherModel.getSignalCode(),currentInfoGatherModel.getSignalType()
                    , currentInfoGatherModel.getSignalName()  ,currentInfoGatherModel.getGatherId(),currentInfoGatherModel.getValue(),currentInfoGatherModel.getValunit(),"0");
        } catch (IOException e) {
            e.printStackTrace();
            XxlJobLogger.log("kafka消费端采集口实体{}转换失败！",record.value().toString(),e.getMessage());
        }
    }

//
//    /**
//     * 处理布尔量
//     * @param record
//     */
//    public void dealBLMessage(ConsumerRecord record){
//        try {
//            CurrentInfoHkModel currentInfoHkModel= JSONUtils.json2pojo(record.value().toString(), CurrentInfoHkModel.class);
//            String retStr = currentInfoHkModel.getValue();
//            int pointAddr =Integer.valueOf(currentInfoHkModel.getPointAddr());
//            String value ="";
//            int retSum = Integer.parseInt(retStr.substring(16, 18), 16);
//            for (int i = 0; i < retSum; i++) {
//                value = retStr.substring(18 + i*2, 18 + (i + 1) * 2);
//                if (value.equals("00")) {
//                    value = "00000000";
//                } else {
//                    value = TransUtils.hexString2binaryString(value);
//                }
//                for (int j = 0; j <value.length() ; j++) {
//                    pointAddr = pointAddr +  1;//地址 地址依次加一
//                    String values=value.charAt(value.length()-1-j)+"";
//                    SignalDefHkModel automaticControlModel = new SignalDefHkModel();
//                    automaticControlModel.setGatherId(currentInfoHkModel.getGatherId());
//                    automaticControlModel.setGatherResId(currentInfoHkModel.getGatherResId());
//                    automaticControlModel.setSignalValue(values);
//                    automaticControlModel.setPointAddr(String.valueOf(pointAddr));//寄存器地址
//                    //    System.out.println("pointAddr: " + String.valueOf(pointAddr)+ " value : " + values+ " str : " + value);
//                    save(automaticControlModel, 0);
//
//                }
//            }
//
//        } catch (IOException e) {
//            e.printStackTrace();
//            XxlJobLogger.log("kafka消费端环控实体{}转换失败！",record.value().toString(),e.getMessage());
//        }
//    }
//    /**
//     * 布尔量保存
//     *
//     * @author wang.kunlin
//     * @date 2020年08月19日 16:56
//     */
//    public void save(SignalDefHkModel automaticControlModel, int flag){
//
//        List<SignalDefHkModel> queryHkSignalList=signalDefService.queryHkSignalList(
//
//                automaticControlModel.getGatherId(), automaticControlModel.getGatherResId(),automaticControlModel.getPointAddr());
//        if(queryHkSignalList==null){
//            // LOGGER.error("自控系统：未找到对应的信号配置");
//            return;
//        }else{
//            for (int num = 0; num < queryHkSignalList.size(); num++) {
//                String nDate="";
//                nDate= ThreadLocalDateUtil.formatDate(new Date());
//
//                String signalType  =queryHkSignalList.get(num).getSignalType();
//
//                //新建实时存储的实体
//                CurrentInfoHkModel currentInfoHkModel=new CurrentInfoHkModel();
//
//                currentInfoHkModel.setValue(automaticControlModel.getSignalValue());
//                currentInfoHkModel.setGatherResId(automaticControlModel.getGatherResId());
//                currentInfoHkModel.setGatherId(automaticControlModel.getGatherId());
//                currentInfoHkModel.setDevCode(queryHkSignalList.get(num).getDevCode());
//                currentInfoHkModel.setSignalCode(queryHkSignalList.get(num).getCode());
//                currentInfoHkModel.setSignalName(queryHkSignalList.get(num).getName());
//                currentInfoHkModel.setSignalType(queryHkSignalList.get(num).getSignalType());
//                currentInfoHkModel.setValunit(queryHkSignalList.get(num).getValunit());
//                currentInfoHkModel.setTime(nDate);
//                currentInfoHkModel.setPointAddr(queryHkSignalList.get(num).getPointAddr());
//
//                //根据实体去redis查询上次信号时间
//                CurrentInfoHkModel currentInfoHkModel1=currentInfoRepository.queryHkCurrent(queryHkSignalList.get(num).getDevCode(),queryHkSignalList.get(num).getCode());
//                int a = 0;
//                if(currentInfoHkModel1==null){
//                    currentInfoHkModel.setLastTime(nDate);
//                }else{
//                    currentInfoHkModel.setLastTime(currentInfoHkModel1.getTime());
//                    if(currentInfoHkModel.getValue().equals(currentInfoHkModel1.getValue())){
//                        a =1;
//                    }
//                }
//                currentInfoHkModel.setSendTime(ThreadLocalDateUtil.formatDate(new Date()));
////                if(){
////
////                }
//                //实时数据存入redis
//                currentInfoRepository.save(currentInfoHkModel);
//
//                //发送kafka
//                send(currentInfoHkModel);
//            }
//
//
//        }
//
//
//
//    }
//
//    /**
//     * 发kafka
//     * @param message
//     */
//    public void send(CurrentInfoHkModel message) {
//
//        try {
//            this.kafkaProduceService.sendHkAlarmMessage(message);
//
//        } catch (Exception e) {
//            XxlJobLogger.log("发送告警信息失败:"+message.toString(), e);
//        }
//    }

}
